package offline;

import generator.CounterGenerator;

import java.io.File;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;

import redis.clients.jedis.SortingParams;
import tlsb.DBException;
import tlsb.Utils;
import util.IFileWriter;
import util.IPartition;
import util.RedisCluster;

/**
 * 由于调用用redis的intersect函数，sourceRedis只能包含一台机器
 * 
 * @author xiafan
 * 
 */
public class CGA {
	final private String n = "n.txt";
	final private String n0 = "n0.txt";
	final private String n1 = "n1.txt";
	final private String n2 = "n2.txt";
	final private String n3 = "n3.txt";
	final private String n4 = "n4.txt";
	final private String n5 = "n5.txt";
	IFileWriter ifr, ifr0, ifr1, ifr2, ifr3, ifr4, ifr5;
	final private String SETOFCOMMUNITIES = ":com";
	final private String PREFIXOFCID = "c:";
	final private String PREFIXOFSID = "s:";
	final private String SETOFOWNEDTL = ":own";
	final private String SETOFBELONGEDTL = ":blg";
	final private String SETOFCURFOLLOWEES = ":cfle";
	final private String SETOFFOLLOWEES = ":fle";
	final private String ALLUIDS = "uids";
	final private String ALLSIDS = "sids";
	final private String READFREQFIELD = "rf";
	final private String WRITEFREQFIELD = "wf";
	final private String MAXCUTSFIELD = "xcuts";
	final private String FSTCOMFIELD = "fstc";
	final private String TODOLIST = "todoList";
	final private String PREFIXOFCIDFORSID = "cfs";
	final private int MAX = 999999999;
	final public int T = 10;
	long counter = 0;
	long sidcounter = 0;
	long memcounter = 0;

	// final private String SETOFMEMBERS = ":mem";

	RedisCluster retRedis = null;
	RedisCluster sourceRedis = null;
	IPartition rPart = null;
	IPartition sPart = null;
	Map<Integer, IFileWriter> ifrMap = new HashMap<Integer, IFileWriter>();

	static Random random = new Random();
	// TODO 导入到单一的节点：207或者208
	public static final String[] HOSTS = { "10.11.1.52", "10.11.1.54",
			"10.11.1.55", "10.11.1.56", "10.11.1.58", "10.11.1.62" };
	public static final int PORT = 6379;
	private boolean _Debug = true;

	CounterGenerator sidsequence;

	public void connect() {
		retRedis = new RedisCluster(new String[] { "10.11.1.208" }, PORT);
		sourceRedis = new RedisCluster(new String[] { "10.11.1.208" }, PORT);
		rPart = retRedis.getPartition();
		sPart = sourceRedis.getPartition();

		ifr = new IFileWriter(new File(n), true);
		ifr.open();
		ifr0 = new IFileWriter(new File(n0), true);
		ifr0.open();
		ifr1 = new IFileWriter(new File(n1), true);
		ifr1.open();
		ifr2 = new IFileWriter(new File(n2), true);
		ifr2.open();
		ifr3 = new IFileWriter(new File(n3), true);
		ifr3.open();
		ifr4 = new IFileWriter(new File(n4), true);
		ifr4.open();
		ifr5 = new IFileWriter(new File(n5), true);
		ifr5.open();

		ifrMap.put(0, ifr0);
		ifrMap.put(1, ifr1);
		ifrMap.put(2, ifr2);
		ifrMap.put(3, ifr3);
		ifrMap.put(4, ifr4);
		ifrMap.put(5, ifr5);
		sidsequence = new CounterGenerator(10000000);

	}

	public void init() throws DBException {
		connect();
		counter = 0;
		// sidsequence = new CounterGenerator(100000);

		// Set<String> uids = jedis.smembers(ALLUIDS);
		int maxcuts = MAX;
		// String sid = "";
		// jedis.del(TODOLIST);

		for (String uid : sourceRedis.getJedis(sPart.shard(ALLUIDS)).smembers(
				ALLUIDS)) {
			// 为每个用户b初始化maxcut
			String thisRF = "";
			// calculate and store the maximum cuts for each user's HTL
			if ((thisRF = sourceRedis.getJedis(sPart.shard(uid)).hget(uid,
					READFREQFIELD)) != null) {
				if (thisRF.contains(".")) {
					thisRF = thisRF.substring(0, thisRF.indexOf("."));
				}
				maxcuts = (int) Math.ceil(T
						/ Math.max(1, Integer.parseInt(thisRF)));
			}
			// if one user has no read frequency, make its maxcuts as 10
			// defaultly
			if (maxcuts == MAX)
				maxcuts = 10;
			if (maxcuts > 1) {
				sourceRedis.getJedis(sPart.shard(uid)).hset(uid, MAXCUTSFIELD,
						Integer.toString(maxcuts));
				sourceRedis.getJedis(sPart.shard(TODOLIST))
						.lpush(TODOLIST, uid);
				// make a copy for original followees set
				for (String member : sourceRedis.getJedis(sPart.shard(uid))
						.smembers(Utils.joinStr(uid, SETOFFOLLOWEES))) {
					sourceRedis.getPipeline(sPart.shard(uid)).sadd(
							Utils.joinStr(uid, SETOFCURFOLLOWEES), member);
				}
				sourceRedis.synAllPipelines();

			} else {
				// 小于1表示timeline不可分割,直接就被丢到shared timeline中去
				Set<String> members = sourceRedis.getJedis(sPart.shard(uid))
						.smembers(Utils.joinStr(uid, SETOFFOLLOWEES));

				String cid = getFirstCommunity(uid);
	
				if (members.size() > 0) {
					String sid = addSTL(members, cid);
					// 001:own s1 s2 record stls one user owns
					retRedis.getJedis(rPart.shard(cid)).sadd(
							Utils.joinStr(uid, SETOFOWNEDTL), sid);
					ifrMap.get(rPart.shard(cid)).writeln(
							"s\t" + Utils.joinStr(uid, SETOFOWNEDTL) + "\t"
									+ sid);

				}
				sourceRedis.getJedis(sPart.shard(uid)).del(
						Utils.joinStr(uid, SETOFFOLLOWEES));
			}
			counter++;
			printcounter(counter, "init");
		}
	}

	public void computeSTLS() {
		counter = 0;
		long listsize = 0;
		while ((listsize = sourceRedis.getJedis(sPart.shard(TODOLIST)).llen(
				TODOLIST)) > 0) {
			System.out.println("@@@@@@@@@@@@@@@" + listsize);
			// sort by cuts
			SortingParams sp = new SortingParams();
			sp.by("*->" + MAXCUTSFIELD);
			List<String> SList = sourceRedis.getJedis(sPart.shard(TODOLIST))
					.sort(TODOLIST, sp);

			String fstUid = SList.get(0);
			String mcf = sourceRedis.getJedis(sPart.shard(fstUid)).hget(fstUid,
					MAXCUTSFIELD);
			int fstCuts = 0;
			if (mcf != null) {
				fstCuts = Integer.parseInt(mcf);
			}
			int curCuts = Math.max(1, fstCuts);
			for (int i = 0; i < SList.size(); i++) {
				String uid = SList.get(i);
				mcf = sourceRedis.getJedis(sPart.shard(uid)).hget(uid,
						MAXCUTSFIELD);
				int thisCuts = 0;
				if (mcf != null) {
					thisCuts = Integer.parseInt(mcf);
				}
				// only the user with the least cuts are processed in one loop
				if (thisCuts > curCuts)
					break;
				thisCuts = intersectWithSTL(uid, thisCuts, null);
				if (thisCuts > 1) {
					// compute stl in related communities
					// 对于curfolloweelist中还有元素并且curcuts>1的情况，与uid所属的cid中的所有成员做交集
					Set<String> cidset = getCommunities(uid);
					List<String> cidList = null;
					if (cidset != null) {
						// 如果在一个小的community中进行处理，不在包含它的更大的community中处理
						cidList = preprocessCommunities(cidset);
					}

					if (cidList.size() > 0) {
						int newCuts = intersectWithComUsers(uid, thisCuts,
								cidList);
						if (newCuts < thisCuts)
							continue;
					}
				}

				moveHTL(uid);
				SList.remove(uid);
				sourceRedis.getJedis(sPart.shard(TODOLIST)).lrem(TODOLIST, 1,
						uid);
				counter++;
				printcounter(counter, "cuts");
			}// end 处理有所maxCuts==curCuts的元素
		}// end 所有用户的关注都已经被切割成STL
	}

	public List<String> preprocessCommunities(Set<String> cidset) {
		List<String> cidList = new LinkedList<String>();
		for (String cid : cidset) {
			int cidcounter = 0;
			cidcounter++;
			if (cidcounter == 1) {
				cidList.add(cid);
				continue;
			}
			if (cidcounter == 2) {
				if (cid.startsWith(cidList.get(0))) {
					cidList.remove(0);
					cidList.add(cid);
					continue;
				}
				if (!cidList.get(0).startsWith(cid)) {
					cidList.add(cid);
				}
			}
			if (cidcounter == 3) {
				for (String ecid : cidList) {
					if (cid.startsWith(ecid)) {
						cidList.remove(ecid);
					}
				}
				boolean flag = true;
				for (String ecid : cidList) {
					if (ecid.startsWith(cid)) {
						flag = false;
					}
				}
				if (flag) {
					cidList.add(cid);
				}
			}
		}
		return cidList;
	}

	/**
	 * 在communities的user中一直进行迭代，知道thisCuts==1。但是目前的实现好像不是这样的，只进行了一次 迭代就结束了 TODO
	 * 性能优化
	 * 
	 * @param uid
	 * @param thisCuts
	 * @param cidList
	 * @return
	 */
	public int intersectWithComUsers(String uid, int thisCuts,
			List<String> cidList) {
		// TODO 是否在所有的community中去找
		long maxOvlp = 0;
		String targetU = "";
		String targetC = "";
		String fstcid = cidList.get(0);
		for (String u : sourceRedis.getJedis(sPart.shard(uid)).smembers(
				Utils.joinStr(PREFIXOFCID, fstcid))) {
			if (u.equals(uid))
				continue;
			if (!sourceRedis.getJedis(sPart.shard(u)).exists(
					Utils.joinStr(u, SETOFCURFOLLOWEES)))
				continue;

			long thisOvlp = 0;
			Set<String> interset = sourceRedis.getJedis(sPart.shard(u)).sinter(
					Utils.joinStr(u, SETOFCURFOLLOWEES),
					Utils.joinStr(uid, SETOFCURFOLLOWEES));

			if (interset != null)
				// 计算交集的写频率
				for (String f : interset) {
					String thisWF = sourceRedis.getJedis(sPart.shard(f)).hget(
							f, WRITEFREQFIELD);
					if (thisWF != null) {
						if (thisWF.contains(".")) {
							thisWF = thisWF.substring(0, thisWF.indexOf("."));
						}
						thisOvlp = +Math.max(0, Integer.parseInt(thisWF));
					}
				}
			if (thisOvlp > maxOvlp) {
				maxOvlp = thisOvlp;
				targetU = u;
				targetC = fstcid;
			}
		}
		// }
		// do when there is a overlapped user
		if (maxOvlp > 0) {
			Set<String> members = sourceRedis.getJedis(sPart.shard(targetU))
					.sinter(Utils.joinStr(targetU, SETOFCURFOLLOWEES),
							Utils.joinStr(uid, SETOFCURFOLLOWEES));
			for (String mem : members) {
				sourceRedis.getPipeline(sPart.shard(uid)).srem(
						Utils.joinStr(uid, SETOFCURFOLLOWEES), mem);
				sourceRedis.getJedis(sPart.shard(targetU)).srem(
						Utils.joinStr(targetU, SETOFCURFOLLOWEES), mem);
			}
			sourceRedis.synAllPipelines();
			// System.err.print("users:"+uid+" "+targetU+" "+members.toString());
			String newsid = addSTL(members, targetC);
			ifrMap.get(rPart.shard(targetC)).writeln(
					"s\t" + Utils.joinStr(uid, SETOFOWNEDTL) + "\t" + newsid);
			ifrMap.get(rPart.shard(getFirstCommunity(targetU))).writeln(
					"s\t" + Utils.joinStr(targetU, SETOFOWNEDTL) + "\t"
							+ newsid);

			retRedis.getJedis(retRedis.getPartition().shard(targetC)).sadd(
					Utils.joinStr(uid, SETOFOWNEDTL), newsid);
			retRedis.getJedis(
					retRedis.getPartition().shard(getFirstCommunity(targetU)))
					.sadd(Utils.joinStr(targetU, SETOFOWNEDTL), newsid);
			sourceRedis.getJedis(sPart.shard(uid)).hincrBy(uid, MAXCUTSFIELD,
					-1);
			thisCuts--;
			sourceRedis.getJedis(sPart.shard(targetU)).hincrBy(targetU,
					MAXCUTSFIELD, -1);
		}
		return thisCuts;
	}

	private static final long OVLP_THRESHOLD = 5;

	/**
	 * 从已有的STL中选出和uid的followee有交际的那些STL，最后返回uid还可以进行分割的curcut
	 * 
	 * @param uid
	 * @param curFollowees
	 * @return
	 */
	public int intersectWithSTL(String uid, int thisCuts,
			Set<String> curFollowees) {
		// do if existed stls can be reshared
		long maxOvlp = OVLP_THRESHOLD;

		Set<String> matchSids = new HashSet<String>();
		// 下面这段代码需要优化，不能够和所有的stl去比
		if (sourceRedis.getJedis(sPart.shard(ALLUIDS)).exists(ALLSIDS)
				&& (sourceRedis.getJedis(sPart.shard(ALLSIDS)).scard(ALLSIDS) != 0)) {
			for (String sid : sourceRedis.getJedis(sPart.shard(ALLSIDS))
					.smembers(ALLSIDS)) {
				long thisOvlp = 0;
				// find out the existing stl that can be reused
				Set<String> diffset = sourceRedis.getJedis(sPart.shard(sid))
						.sdiff(Utils.joinStr(PREFIXOFSID, sid),
								Utils.joinStr(uid, SETOFFOLLOWEES));
				if (diffset == null || diffset.size() <= 0) {
					matchSids.add(sid);
				}
			}
		}
		do {
			// TODO不用每次读取subset list
			if (thisCuts <= 1) {
				break;
			} else {
				maxOvlp = OVLP_THRESHOLD;
				String reusedSid = "";

				for (String sid : matchSids) {
					long thisOvlp = 0;
					// find out the existing stl that can be reused
					for (String u : sourceRedis.getJedis(sPart.shard(sid))
							.sinter(Utils.joinStr(PREFIXOFSID, sid),
									Utils.joinStr(uid, SETOFCURFOLLOWEES))) {
						String thisWF = sourceRedis.getJedis(sPart.shard(u))
								.hget(u, WRITEFREQFIELD);
						if (thisWF != null) {
							if (thisWF.contains(".")) {
								thisWF = thisWF.substring(0,
										thisWF.indexOf("."));
							}
							thisOvlp = +Math.max(0, Integer.parseInt(thisWF)); // 计算和当前用户交集的写频率
						}
					}
					if (thisOvlp > maxOvlp) {
						maxOvlp = thisOvlp;
						reusedSid = sid;
					}
				}
				// do when there's reused stl
				if (maxOvlp > OVLP_THRESHOLD) {
					matchSids.remove(reusedSid);
					thisCuts++;
					for (String mem : sourceRedis.getJedis(
							sPart.shard(reusedSid)).smembers(
							Utils.joinStr(PREFIXOFSID, reusedSid))) {
						sourceRedis.getPipeline(sPart.shard(uid)).srem(
								Utils.joinStr(uid, SETOFCURFOLLOWEES), mem);
					}
					sourceRedis.getPipeline(sPart.shard(uid)).hincrBy(uid,
							MAXCUTSFIELD, -1);
					sourceRedis.synAllPipelines();

					// TODO record the reused stl for uid
					String cid = getFirstCommunity(uid);

					retRedis.getJedis(rPart.shard(cid)).sadd(
							Utils.joinStr(uid, SETOFOWNEDTL), reusedSid);

					ifrMap.get(rPart.shard(cid)).writeln(
							"s\t" + Utils.joinStr(uid, SETOFOWNEDTL) + "\t"
									+ reusedSid);
				}
			}
		} while (maxOvlp > OVLP_THRESHOLD); // 在subset
											// list中的subset与curfolloweelist没有交集
		return thisCuts;
	}

	/**
	 * members构成一个share timeline，cid是该share timeline对应的cid
	 * 
	 * @param members
	 * @param cid
	 * @return
	 */
	public String addSTL(Set<String> members, String cid) {
		String sid = sourceRedis.getJedis(sPart.shard(members.toString())).get(
				Utils.joinStr("X:", members.toString()));
		if (sid != null) {
			return sid;
		} else {
			// 插入的share timeline不存在时的逻辑
			sid = Utils.hash(sidsequence.nextInt()) + "";
			sourceRedis.getJedis(sPart.shard(ALLSIDS)).sadd(ALLSIDS, sid);
			sourceRedis.getJedis(sPart.shard(sid)).set(
					Utils.joinStr(PREFIXOFCIDFORSID, sid), cid); // 存储sid的cid
			ifr.writeln(Utils.joinStr(PREFIXOFCIDFORSID, sid) + "\t" + cid);

			sidcounter++;
			if (sidcounter % 100000 == 0) {
				System.out.println(sidcounter + "sids added");
			}

			// fle的status需要push到sid
			for (String fle : members) {
				memcounter++;
				String flecid = getFirstCommunity(fle);
				// s:001 f1 f2 record uid in an stl
				sourceRedis.getPipeline(sPart.shard((sid))).sadd(
						Utils.joinStr(PREFIXOFSID, sid), fle);
				// 001:blg s1 s2 record stls one user belongs to

				if (flecid == null) {
					System.err.println("move has no fst com:" + fle);
					flecid = random.nextInt(9999999) + "";
					sourceRedis.getPipeline(sPart.shard(fle)).hset(fle,
							FSTCOMFIELD, flecid);
					ifr.writeln(FSTCOMFIELD + "\t" + fle + "\t" + flecid);
				}
				int nid = rPart.shard(flecid);

				retRedis.getPipeline(nid).sadd(
						Utils.joinStr(fle, SETOFBELONGEDTL), sid);
				ifrMap.get(nid).writeln(
						"s\t" + Utils.joinStr(fle, SETOFBELONGEDTL) + "\t"
								+ sid);

			}
			sourceRedis.synAllPipelines();
			retRedis.synAllPipelines();
			String smembers = members.toString();

			// 以sid中的uids为key记录
			sourceRedis.getJedis(sPart.shard(smembers)).set(
					Utils.joinStr("X:", smembers), sid);
			// System.err.println(sid+":"+smembers);
			return sid;
		}
	}

	// 把uid的curfollowees作为一个share timeline处理
	public void moveHTL(String uid) {
		if (sourceRedis.getJedis(sPart.shard(uid)).scard(
				Utils.joinStr(uid, SETOFCURFOLLOWEES)) <= 1) {
			sourceRedis.getJedis(sPart.shard(TODOLIST)).lrem(TODOLIST, 1, uid);
			counter++;
			printcounter(counter, "cuts");
		}

		Set<String> members = sourceRedis.getJedis(sPart.shard(uid)).smembers(
				Utils.joinStr(uid, SETOFCURFOLLOWEES));

		String cid = getFirstCommunity(uid);
	
		if (members.size() > 0) {
			String sid = addSTL(members, cid);
			// 001:own s1 s2 record stls one user owns
			ifrMap.get(rPart.shard(cid)).writeln(
					"s\t" + Utils.joinStr(uid, SETOFOWNEDTL) + "\t" + sid);

			retRedis.getJedis(rPart.shard(cid)).sadd(
					Utils.joinStr(uid, SETOFOWNEDTL), sid);
		}
		sourceRedis.getJedis(sPart.shard(uid)).del(
				Utils.joinStr(uid, SETOFCURFOLLOWEES));
	}

	public void cleanup() throws DBException {
		System.out.println("all members in" + sidcounter + " sid:" + counter);

		ifr.close();
		ifr0.close();
		ifr1.close();
		ifr2.close();
		ifr3.close();
		ifr4.close();
		ifr5.close();
	}

	public String getFirstCommunity(String uid) {
		String cid = sourceRedis.getJedis(sPart.shard(uid)).hget(uid,
				FSTCOMFIELD);
		if (cid == null) {
			cid = random.nextInt(9999999) + "";
			setFirstCommunity(uid, cid);
			ifr.writeln(FSTCOMFIELD + "\t" + uid + "\t" + cid);
			System.err.println("no fst com");
		}
		return cid;
	}

	public void setFirstCommunity(String uid, String cid) {
		sourceRedis.getJedis(sPart.shard(uid)).hset(uid, FSTCOMFIELD, cid);
	}

	public Set<String> getCommunities(String uid) {
		return sourceRedis.getJedis(sPart.shard(uid)).smembers(
				Utils.joinStr(uid, SETOFCOMMUNITIES));
	}

	public void printcounter(long counter2, String comment) {
		// counter2++;
		if (comment.equals("cuts") && counter2 % 10000 == 0)
			System.out.println(comment + ":" + counter2);
		else if (counter2 % 100000 == 0)
			System.out.println(comment + ":" + counter2);

	}

	public static void main(String[] args) throws Exception {

		CGA cga = new CGA();

		cga.connect();

		cga.init();

		cga.computeSTLS();

		cga.cleanup();

	}

}
